package com.zookeeper.demo1.lock;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

public class DistributedLock {

    private String tmpPath = "/lock_distributed";

    private String seqPath = "lock_";

    private Function<String, Integer> integerFunction = Integer::valueOf;

    private ZooKeeper zooKeeper;

    private static ThreadLocal<Boolean> threadLocal = new ThreadLocal<>();


    public void init() throws IOException {
        zooKeeper = new ZooKeeper("192.168.93.10:2181", 60 * 1000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {

            }
        });
    }

    public void delete(String path) throws Exception {
        zooKeeper.delete(path, -1);
    }


    public String getLock(int time) throws Exception {
        String path = null;
        //1.创建顺序临时节点
        String createPath = tmpPath + "/" + seqPath;
        String seqNode = zooKeeper.create(createPath, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        int createSeq = integerFunction.apply(seqNode.substring(23));
        //2.获取所有的顺序节点
        List<String> children = zooKeeper.getChildren(tmpPath, false);
        System.out.println(children);
        Map<Integer, String> map = children.stream().map(i -> {
            Integer seq = integerFunction.apply(i.substring(5));
            return new Node(seq, i);
        }).collect(Collectors.toMap(Node::getSeq, Node::getPath, (k, v) -> v));
        List<Integer> collect = map.keySet().
                stream().
                sorted((a, b) -> a.compareTo(b)).
                collect(Collectors.toList());
        //取出最小的那个临时节点
        Integer minSeq = collect.get(0);
        String minPath = map.get(minSeq);
        if (minSeq != createSeq) {
            LockStatus.STATUS state = getWaitLock(tmpPath + "/" + minPath, createSeq, time).getState();
            System.out.println(state);
            if (state == LockStatus.STATUS.FAILURE) {
                //将之前创建的节点删除
                delete(seqNode);
                return null;
            }

        }
        path = seqNode;
        threadLocal.remove();
        return path;
    }


    private LockStatus getWaitLock(String minPath, Integer createSeq, int timeout) throws KeeperException, InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);
        LockStatus status = new LockStatus();
        threadLocal.set(Boolean.FALSE);
        //3.没有获取到锁，监控最小的节点的删除事件
        zooKeeper.exists(minPath, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (Event.EventType.NodeDeleted == event.getType()) {
                    //删除事件
                    try {
                        while (true) {
                            //这里的判断其实就是为了如果监听到时间，但是已经超过客户端设置的超时时间，因为到这里也不一定能够获取到锁，所以直接放弃
                            if (threadLocal.get() == Boolean.TRUE) {
                                threadLocal.remove();
                                status.setState(LockStatus.STATUS.FAILURE);
                                break;
                            }
                            List<String> children = zooKeeper.getChildren(tmpPath, false);
                            Map<Integer, String> map = children.stream().map(i -> {
                                Integer seq = integerFunction.apply(i.substring(5));
                                return new Node(seq, i);
                            }).collect(Collectors.toMap(Node::getSeq, Node::getPath, (k, v) -> v));
                            List<Integer> collect = map.keySet().
                                    stream().
                                    sorted((a, b) -> a.compareTo(b)).
                                    collect(Collectors.toList());
                            Integer minSeq = collect.get(0);
                            if (minSeq == createSeq) {
                                latch.countDown();
                                status.setState(LockStatus.STATUS.SUUCESS);
                                break;
                            }
                            if (minSeq != createSeq) {
                                //没有获取到锁，先让出cpu的执行权限
                                Thread.yield();
                            }
                        }

                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
            }
        });
        if (timeout > 0) {
            //等待超时的时间内没有获取到锁就放弃获取锁
            threadLocal.set(latch.await(timeout, TimeUnit.SECONDS));
        } else {
            latch.await();
        }
        return status;
    }
}
